/**
 * 
 */
package io.dora.registry.curator;

import io.dora.conf.ServiceURL;
import io.dora.registry.AbstractRegistry;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Curator zookeeper服务注册中心实现
 * 
 * @author zhaohongbo
 *
 */
public class CuratorRegistry extends AbstractRegistry {
	
	private Logger logger = LoggerFactory.getLogger(CuratorRegistry.class);
	
	private CuratorFramework zkClient;
	
	public CuratorRegistry(String url) {
		zkClient = CuratorFrameworkFactory
				.builder()
				.connectString(url)
				.retryPolicy(new RetryNTimes(2000,20000))
				.build();
		zkClient.start();
	}

	@Override
	public void register(ServiceURL serviceURL) {
		boolean flag = CuratorUtils.createNode(zkClient, serviceURL.getPath());
		if (flag) {
			CuratorUtils.addOrUpdateNodeData(zkClient, serviceURL.getPath(), serviceURL.getHost() + ":" + serviceURL.getPort(), false);
		}
	}

	@Override
	public void lookup(String serviceKey) {
		String data = CuratorUtils.getNodeData(zkClient, serviceKey);
		if (StringUtils.isNotBlank(data)) {
			serviceMap.put(serviceKey, data);
			this.subscribe(serviceKey);
			logger.info("Lookup service by key: {}", serviceKey);
			logger.info("Lookup service hosts: {}", data);
		}
	}

	@Override
	public void subscribe(final String serviceKey) {
		final NodeCache nodeCache = new NodeCache(zkClient, serviceKey);
		try {
			nodeCache.start(true);
		} catch (Exception e) {
			logger.error("{}", e);
		}
		nodeCache.getListenable().addListener(new NodeCacheListener () {
			@Override
			public void nodeChanged() {
				String data = CuratorUtils.getNodeData(zkClient, serviceKey);
				serviceMap.put(serviceKey, data);
				logger.info("{} new data {}", serviceKey, data);
			}
		}, pool);
	}

}
